#!/usr/bin/env python
# -*- coding: UTF-8 -*-
'''
@Project ：V2 
@File    ：ga_fjsp_oop.py
@IDE     ：PyCharm 
@Author  ：郭星
@Date    ：2025/9/20 21:49 
'''
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""

帮我写一个简单的遗传算法求解柔性作业车间调度问题，最好能按照类的角度去写 机器类 工件类这种 写的系统且模块
ChatGPT 说：
已思考 8s

没问题！下面是一份简洁但模块化（面向对象）的 Python 单文件代码，用遗传算法（GA）求解柔性作业车间调度（FJSP）。
代码把问题拆成工件/工序/机器/问题/解码器/GA六类，结构清晰，便于后续扩展（比如加能耗、加换型、加动态等）。
说明与扩展

编码：

op_seq 用“重复作业 id 序列”，确保每个作业恰好出现其工序数次。

mac_seq 存的是“可行机列表的索引”，避免直接映射到全局机器号导致不可行。

解码：FJSPDecoder.decode() 严格遵守作业工序顺序与机器不可重叠，计算 start = max(job_ready, machine_ready)，得到 Cmax 与机器甘特。

算法：

交叉：op_seq 用 PPX（保持先后关系和出现次数），mac_seq 用均匀交叉。

变异：op_seq 随机交换，mac_seq 按当前 op_seq 的第 i 位对应的“工序”重采样可行机索引。

如何换成你的实例：把 build_random_fjsp() 换成 CSV/JSON 解析，生成 jobs: List[Job] 即可；其它模块不变。

下一步可以加：

多目标（Cmax + total tardiness）→ NSGA-II 结构；

setup/换型：在 Machine.schedule() 前增加换型时间计算；

动态（释放时间、停机）→ 参考我之前的“动态+滚动重调度”版本，用本 GA 做滚动优化器；

并行/多进程评估；更强初始化（NEH-like 启发式）等。

如果你希望我把数据输入/输出接口（CSV/Excel + 结果导出）再封装一下，也可以在这套类的基础上继续
GA for Flexible Job Shop Scheduling (FJSP) - OOP version
--------------------------------------------------------
- Problem entities: Operation, Job, Machine, FJSPProblem
- Decoder: from chromosome (op_seq + mac_seq) -> feasible schedule & makespan
- GA: PPX for operation sequence, uniform crossover for machine choices, simple mutations
- Demo: random instance generator + convergence curve + optional Gantt chart

Author: you
"""

import random
import numpy as np
from dataclasses import dataclass
from typing import Dict, List, Tuple, Optional

try:
    import matplotlib.pyplot as plt
    HAS_MPL = True
except Exception:
    HAS_MPL = False


# ========== 基础实体类 ==========

@dataclass
class Operation:
    """
    一道工序：可在若干机器上加工，每台机器有一个对应的加工时间
    machine_time: {machine_id: processing_time}
    """
    machine_time: Dict[int, int]


@dataclass
class Job:
    """
    一个工件：由若干道顺序执行的工序组成
    ops: List[Operation]
    """
    ops: List[Operation]

    @property
    def n_ops(self) -> int:
        return len(self.ops)


@dataclass
class Task:
    """用于机器时间轴记录"""
    job_id: int
    op_idx: int
    start: float
    end: float

    @property
    def duration(self) -> float:
        return self.end - self.start


class Machine:
    """
    机器：维护自己的加工时间轴（不可抢占），并提供调度记录接口
    """
    def __init__(self, machine_id: int):
        self.id = machine_id
        self.timeline: List[Task] = []
        self.ready_time: float = 0.0

    def schedule(self, job_id: int, op_idx: int, start: float, pt: float):
        st = max(self.ready_time, start)
        en = st + pt
        self.timeline.append(Task(job_id, op_idx, st, en))
        self.ready_time = en
        return st, en


class FJSPProblem:
    """
    柔性作业车间问题定义：
    - jobs: List[Job]
    - m: 机器数
    """
    def __init__(self, jobs: List[Job], m: int):
        self.jobs = jobs
        self.m = m

    @property
    def n_jobs(self) -> int:
        return len(self.jobs)

    @property
    def n_ops_total(self) -> int:
        return sum(job.n_ops for job in self.jobs)

    def feasible_machines(self, j: int, k: int) -> List[int]:
        """返回工件 j 的第 k 道工序的可行机器列表"""
        return list(self.jobs[j].ops[k].machine_time.keys())


# ========== 解码器：从染色体到日程与 Cmax ==========

class FJSPDecoder:
    """
    染色体编码：
      - op_seq: 长度 = 总工序数，由作业 id 组成（每个作业重复出现其工序数次），表示工序出现顺序
      - mac_seq: 长度 = 总工序数，每次出现（某作业的下一个待加工工序）选用的机器在“可行机器列表”的索引（而非全局机器号）
    解码步骤：
      - 记录每个作业“下一道工序索引”和“就绪时间”，每台机器“就绪时间”
      - 遍历 op_seq 的每个位置：确定该作业的第 k 道工序，按照 mac_seq 选择可行机中的某一台
      - 计算开始时间 = max(作业就绪, 机器就绪)，结束时间 = 开始 + pt，更新状态与时间轴
    """
    def __init__(self, problem: FJSPProblem):
        self.pb = problem

    def decode(self, op_seq: List[int], mac_seq: List[int]) -> Tuple[float, List[Machine]]:
        # 初始化状态
        job_next = np.zeros(self.pb.n_jobs, dtype=int)     # 每个作业的下一个待加工工序号
        job_ready = np.zeros(self.pb.n_jobs, dtype=float)  # 每个作业的就绪时间（上一道完成）
        machines = [Machine(i) for i in range(self.pb.m)]  # 机器时间轴

        mac_ptr = 0
        for j in op_seq:
            k = int(job_next[j])
            if k >= self.pb.jobs[j].n_ops:
                # 该作业已完成（保护性判断），跳过
                continue

            feas = self.pb.feasible_machines(j, k)
            if len(feas) == 0:
                # 没有可行机（异常输入）
                continue

            # 从 mac_seq 取一个下标映射到可行机器 id
            idx = mac_seq[mac_ptr] % len(feas)
            mm = int(feas[idx])
            pt = float(self.pb.jobs[j].ops[k].machine_time[mm])

            # 安排到机器 mm 上
            start = max(job_ready[j], machines[mm].ready_time)
            st, en = machines[mm].schedule(j, k, start, pt)

            # 更新作业状态
            job_ready[j] = en
            job_next[j] += 1
            mac_ptr += 1

        makespan = float(np.max(job_ready)) if len(job_ready) > 0 else 0.0
        return makespan, machines


# ========== GA：遗传算法 ==========

class GAFJSP:
    """
    简单 GA：
      - 初始化：随机打乱 op_seq；mac_seq 对应位置随机选可行机下标
      - 交叉：
          * op_seq 用 PPX（保持每个作业出现次数）
          * mac_seq 用均匀交叉
      - 变异：
          * op_seq 交换两个位置
          * mac_seq 按一定概率重采样为新的“可行机下标”
    """
    def __init__(self, problem: FJSPProblem,
                 pop: int = 80, gens: int = 400, pc: float = 0.9, pm: float = 0.2,
                 seed: Optional[int] = 0):
        self.pb = problem
        self.pop = pop
        self.gens = gens
        self.pc = pc
        self.pm = pm
        if seed is not None:
            random.seed(seed)
            np.random.seed(seed)
        self.decoder = FJSPDecoder(problem)
        # 基础 op 序列：作业 j 重复出现其工序数次
        self.base_seq: List[int] = []
        for j, job in enumerate(self.pb.jobs):
            self.base_seq += [j] * job.n_ops

    # ---------- 染色体构造 ----------
    def random_individual(self) -> Tuple[List[int], List[int]]:
        # op_seq
        op_seq = self.base_seq[:]
        random.shuffle(op_seq)
        # mac_seq：逐位置根据当前位置对应的“作业的第 k 道工序”的可行机数量随机生成索引
        mac_seq = []
        seen = [0] * self.pb.n_jobs
        for j in op_seq:
            k = seen[j]
            feas = self.pb.feasible_machines(j, k)
            mac_seq.append(random.randrange(max(1, len(feas))))
            seen[j] += 1
        return op_seq, mac_seq

    # ---------- 交叉 ----------
    def crossover_ppx(self, a: List[int], b: List[int]) -> List[int]:
        """PPX-like：保持每个作业出现次数"""
        n = len(a)
        # 统计每个作业需要次数（由a决定即可，因为a与b是同一个基底的排列）
        need = {}
        for x in a:
            need[x] = need.get(x, 0) + 1
        left = dict(need)
        res = []
        ai = bi = 0
        while len(res) < n:
            take_a = (random.random() < 0.5)
            if take_a:
                while ai < n and left.get(a[ai], 0) == 0:
                    ai += 1
                if ai < n:
                    j = a[ai]; res.append(j); left[j] -= 1; ai += 1
                else:
                    while bi < n and left.get(b[bi], 0) == 0:
                        bi += 1
                    j = b[bi]; res.append(j); left[j] -= 1; bi += 1
            else:
                while bi < n and left.get(b[bi], 0) == 0:
                    bi += 1
                if bi < n:
                    j = b[bi]; res.append(j); left[j] -= 1; bi += 1
                else:
                    while ai < n and left.get(a[ai], 0) == 0:
                        ai += 1
                    j = a[ai]; res.append(j); left[j] -= 1; ai += 1
        return res

    def crossover_uniform(self, a: List[int], b: List[int]) -> List[int]:
        """mac_seq 的均匀交叉"""
        return [(aa if random.random() < 0.5 else bb) for aa, bb in zip(a, b)]

    # ---------- 变异 ----------
    def mutate(self, op_seq: List[int], mac_seq: List[int]):
        # 1) op_seq 交换
        if len(op_seq) >= 2 and random.random() < 0.5:
            i, j = random.sample(range(len(op_seq)), 2)
            op_seq[i], op_seq[j] = op_seq[j], op_seq[i]

        # 2) mac_seq 局部重采样（必须根据 op_seq 上的位置映射到对应工序）
        if random.random() < 0.7:
            seen = [0] * self.pb.n_jobs
            for i, j in enumerate(op_seq):
                k = seen[j]
                seen[j] += 1
                if random.random() < 0.2:
                    feas = self.pb.feasible_machines(j, k)
                    mac_seq[i] = random.randrange(max(1, len(feas)))

    # ---------- 适应度 ----------
    def fitness(self, ind: Tuple[List[int], List[int]]) -> float:
        op_seq, mac_seq = ind
        mk, _ = self.decoder.decode(op_seq, mac_seq)
        return mk

    # ---------- 进化 ----------
    def run(self, verbose: bool = True) -> Tuple[Tuple[List[int], List[int]], float, List[float]]:
        # 初始化种群
        pop = [self.random_individual() for _ in range(self.pop)]
        fits = [self.fitness(ind) for ind in pop]

        best_hist = []
        for g in range(1, self.gens + 1):
            new_pop = []
            # 锦标赛 + 交叉 + 变异
            while len(new_pop) < self.pop:
                # 选双亲
                i, j = random.randrange(self.pop), random.randrange(self.pop)
                p1 = pop[i] if fits[i] < fits[j] else pop[j]
                i, j = random.randrange(self.pop), random.randrange(self.pop)
                p2 = pop[i] if fits[i] < fits[j] else pop[j]

                c1 = (p1[0][:], p1[1][:])
                c2 = (p2[0][:], p2[1][:])

                # 交叉
                if random.random() < self.pc:
                    c1 = (self.crossover_ppx(p1[0], p2[0]),
                          self.crossover_uniform(p1[1], p2[1]))
                    c2 = (self.crossover_ppx(p2[0], p1[0]),
                          self.crossover_uniform(p2[1], p1[1]))

                # 变异
                if random.random() < self.pm:
                    self.mutate(*c1)
                if random.random() < self.pm:
                    self.mutate(*c2)

                new_pop += [c1, c2]

            pop = new_pop[:self.pop]
            fits = [self.fitness(ind) for ind in pop]

            best = float(min(fits))
            best_hist.append(best)
            if verbose and g % 50 == 0:
                print(f"[GA] Gen {g:4d}  Best Cmax = {best:.2f}")

        # 返回最优个体与收敛曲线
        idx = int(np.argmin(fits))
        return pop[idx], fits[idx], best_hist


# ========== 可视化（甘特图 & 收敛曲线） ==========

def plot_convergence(best_hist: List[float], title: str = "Convergence"):
    if not HAS_MPL:
        print("(matplotlib not installed; skip convergence plot)")
        return
    import matplotlib.pyplot as plt
    plt.figure(figsize=(6, 4))
    plt.plot(best_hist, lw=2)
    plt.xlabel("Generation")
    plt.ylabel("Best makespan")
    plt.title(title)
    plt.grid(True, alpha=0.3)
    plt.tight_layout()
    plt.show()

def plot_gantt(machines: List[Machine], title: str = "Gantt Chart"):
    if not HAS_MPL:
        print("(matplotlib not installed; skip Gantt plot)")
        return
    import matplotlib.pyplot as plt
    fig, ax = plt.subplots(figsize=(12, 0.8 * len(machines) + 3))
    yticks, ylabels = [], []
    for idx, mach in enumerate(machines):
        for t in sorted(mach.timeline, key=lambda x: x.start):
            ax.barh(idx, t.duration, left=t.start, align='center')
            ax.text(t.start + t.duration / 2, idx, f"J{t.job_id}-O{t.op_idx}",
                    ha='center', va='center', fontsize=8, color='white')
        yticks.append(idx); ylabels.append(f"M{mach.id}")
    ax.set_yticks(yticks); ax.set_yticklabels(ylabels)
    ax.set_xlabel("Time"); ax.set_title(title)
    ax.grid(axis="x", alpha=0.3)
    plt.tight_layout(); plt.show()


# ========== 随机实例构造 ==========

def build_random_fjsp(n_jobs: int = 12, m: int = 6,
                      ops_per_job: int = 4, k_machines_per_op: int = 2,
                      pt_low: int = 1, pt_high: int = 9,
                      seed: Optional[int] = 0) -> FJSPProblem:
    """
    生成一个小型 FJSP 随机实例：
    - 每个工序在 k_machines_per_op 台机器上可加工，时间均匀采样 [pt_low, pt_high]
    """
    rng = np.random.RandomState(seed)
    jobs: List[Job] = []
    for _ in range(n_jobs):
        ops: List[Operation] = []
        for _o in range(ops_per_job):
            feas = sorted(rng.choice(range(m), size=k_machines_per_op, replace=False))
            mt = {mm: int(rng.randint(pt_low, pt_high + 1)) for mm in feas}
            ops.append(Operation(machine_time=mt))
        jobs.append(Job(ops=ops))
    return FJSPProblem(jobs, m)


# ========== Demo 主程序 ==========

def main():
    # 固定随机种子便于复现
    random.seed(0); np.random.seed(0)

    # 1) 构建随机实例
    pb = build_random_fjsp(n_jobs=12, m=6, ops_per_job=4, k_machines_per_op=2, seed=42)

    # 2) 运行 GA
    ga = GAFJSP(pb, pop=90, gens=400, pc=0.9, pm=0.25, seed=1)
    (best_op_seq, best_mac_seq), best_mk, hist = ga.run(verbose=True)
    print(f"\n[Result] Best makespan = {best_mk:.2f}")

    # 3) 最优解解码，拿到机器时间轴
    decoder = FJSPDecoder(pb)
    mk, machines = decoder.decode(best_op_seq, best_mac_seq)
    assert abs(mk - best_mk) < 1e-6

    # 4) 可视化
    plot_convergence(hist, title="FJSP - GA Convergence")
    plot_gantt(machines, title=f"FJSP GA Gantt (Cmax={best_mk:.1f})")

if __name__ == "__main__":
    main()
